package cn.tedu.consumer;

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import java.util.Locale;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * 将消费者交给Spring，添加Spring容器中
 * 实现RocketMQListener
 *  CONSUME_SUCCESS  当前方法是否抛出异常，如果没有异常抛出，表示消费成功
 *  RECONSUME_LATER 代表有异常抛出，表示稍后处理,会重新进行消息消费
 */
@Component
@RocketMQMessageListener(
        topic = "test-01-topic",
        selectorExpression = "*",
        consumerGroup = "${rocketmq.consumer.group}"
)
public class MyConsumer implements RocketMQListener<String> {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public void onMessage(String message) {

        //模拟订单消费，提供一个订单编号
        //1.将抢锁的所有参数准备好
        //1.1 锁的key值：订单编号
        String lockKey="order:add:"+message;
        //1.2 锁的value值
        String randCode = new Random().nextInt(9000) + 1000 + "";
        //2.抢锁
        ValueOperations<String, String> opsForValue = stringRedisTemplate.opsForValue();
        Boolean tryLock = opsForValue.setIfAbsent(lockKey, randCode, 10, TimeUnit.SECONDS);
        try{
            while (!tryLock){
                Thread.sleep(5000);
                tryLock = opsForValue.setIfAbsent(lockKey, randCode, 10, TimeUnit.SECONDS);
            }
            //执行业务逻辑
            //orderService.addOrder();
        }catch (Exception e){

        }finally {
            //手动释放锁
            String value = opsForValue.get(tryLock);
            if (randCode.equals(value)){
                stringRedisTemplate.delete(lockKey);
            }
        }


        System.out.println("将消息消费，消费内容："+message);
        throw new RuntimeException("消息消费失败");
    }
}
